package com.shujia.core

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo24BroadCast {
  def main(args: Array[String]): Unit = {
    /**
     * 一般的join操作是需要reduce任务的
     * mapJoin：只需要map任务就可以实现的join操作
     * 适用于大表关联小表
     * 原理：将小表进行广播，广播给每一个map任务
     */

    // 统计学生总分：id,name,clazz,sumScore

    val conf: SparkConf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("Demo24BroadCast")

    val sc: SparkContext = new SparkContext(conf)

    val stuRDD: RDD[String] = sc
      .textFile("spark/data/stu/students.txt")

    val scoreRDD: RDD[String] = sc
      .textFile("spark/data/stu/score.txt")

    // 计算每个学生的总分
    val sumScoreRDD: RDD[(String, Int)] = scoreRDD
      .map(line => {
        val splits: Array[String] = line.split(",")
        (splits(0), splits(2).toInt)
      }).reduceByKey(_ + _)

    // 以学生id作为key，总分作为value，将RDD转换成Map集合
    val sumScoreMap: Map[String, Int] = sumScoreRDD.collect().toMap
    // 将变量进行广播，广播给每一个Executor
    // 当Task的数量过多时，可以大大降低变量被复制的次数，进而提升性能
    // 前提条件：Executor数量 << Task数量
    val sumScoreMapBro: Broadcast[Map[String, Int]] = sc.broadcast(sumScoreMap)

    // 遍历学生的数据，通过id结合Map的getOrElse方法获取学生总分，实现关联
    stuRDD.map(line => {
      val splits: Array[String] = line.split(",")
      val id: String = splits(0)
      // RDD是不支持被序列化的，所有Task内部是无法使用RDD的，RDD与RDD之间不能嵌套
      //      val sumScore: Int = sumScoreRDD.collect().toMap.getOrElse(id, 0)
      // 如果直接使用外部变量 会导致每个Task中都会有一份变量的副本 如果Task数量比较多会影响性能
      //      val sumScore: Int = sumScoreMap.getOrElse(id, 0)

      val sumScore: Int = sumScoreMapBro.value.getOrElse(id, 0)
      s"$id,${splits(1)},${splits(4)},$sumScore"
    }).foreach(println)

  }

}
